package sunyu.tools.hbase.rowmapper;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.data.hadoop.hbase.RowMapper;
import sunyu.tools.hbase.pojo.Family;
import sunyu.tools.hbase.pojo.Row;

import java.util.Map;

public class HbaseRowMapper implements RowMapper<Row> {

    /**
     * result转换成row对象
     *
     * @param result
     * @param returnRowKeyOnly row对象里面只有rowKey
     *
     * @return
     */
    public static Row mapRow(Result result,
                             Boolean returnRowKeyOnly) {
        if (result == null || result.getRow() == null) {
            return null;
        }
        if (returnRowKeyOnly == null) {
            returnRowKeyOnly = false;
        }
        Row row = new Row();// 一行
        row.setRowName(Bytes.toString(result.getRow()));
        if (returnRowKeyOnly != null && returnRowKeyOnly == false) {
            Map<String, Family> familys = row.getFamilys();// 一行里的family集合
            for (Cell cell : result.rawCells()) {
                String familyName = Bytes.toString(CellUtil.cloneFamily(cell));// 获取当前列的familyName
                String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));// 列名
                String value = Bytes.toString(CellUtil.cloneValue(cell));// 值

                Family family = familys.get(familyName);
                if (family == null) {
                    family = new Family();
                    familys.put(familyName, family);
                }

                Map<String, Object> columns = family.getColumns();// 当前family的所有列
                columns.put(columnName, value);
            }
        }
        return row;
    }

    @Override
    public Row mapRow(Result result,
                      int rowNum) throws Exception {
        return mapRow(result, false);
    }

}
